﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Psm.Messaging;
using Psm.PubSub.Configuration;
using Psm.PubSub.Messaging;
using System.Threading;
using Psm.Logging;
using Psm.Container;
using System.Diagnostics;
using Psm.Session;
using System.Threading.Tasks;

namespace Psm.PubSub.StateCache
{
    public class TopicSpace
    {
        private static ILogger logger = Repository.GetInstance<ILogger>();
        public static int DEFAULT_CAPACITY = 5000000;

        public string Name { get { return this.Schema.Topic; } }
        public int Capacity { get; private set; }
        private DataMessageDecorator[] messageList = null;
        private int count = 0;
        public int Count { get { return count; } }
        public Schema Schema { get; private set; }
        private SpinLock addLock = new SpinLock(); //might need to be changed into simple lock....
        private IList<QueryMessageDecorator> _queryList = new List<QueryMessageDecorator>(100);


        #region PERFORMANCE_CODE
        private Stopwatch watch = new Stopwatch();
        private int TOTALMESSAGECOUNT = 1000000;
        #endregion

        public TopicSpace(Schema schema)
        {
            this.Schema = schema;
            this.Capacity = DEFAULT_CAPACITY;
            messageList = new DataMessageDecorator[this.Capacity];
        }
        
        public int AddMessage(DataMessageInternal newMessage)
        {   
            bool taken = false;
            int index = -1;
            QueryMessageDecorator[] queries = null;
            DataMessageDecorator decorator = new DataMessageDecorator();

            newMessage.ClearPayload();

            //add new record
            addLock.Enter(ref taken);
            //lock (this)
            {
                if (this.count != DEFAULT_CAPACITY)
                {
                    index = count;
                    decorator.SetMessage(newMessage);
                    messageList[index] = decorator;
                    count++;
                }
                if (_queryList.Count > 0)
                    queries = _queryList.ToArray(); //performance?
            }
            addLock.Exit(true);

            //evaluate already registered queries
            //TODO:bug flush issue
            if( queries != null )
                for (int qIndex = 0; qIndex < queries.Length; qIndex++)
                    if (index != -1) messageList[index].EvaluateQuery(queries[qIndex]);


            #region PERFORMANCE_CODE
            if (index == -1) logger.Error("Topic {0} capacity is full!!!", this.Name);
            if (count == 1) watch.Start();
            if (count == TOTALMESSAGECOUNT) 
            { 
                watch.Stop(); 
                logger.Info("{1} messages in {0} for topic {2}", watch.ElapsedMilliseconds, TOTALMESSAGECOUNT, this.Name);
                GC.Collect();
            }
            #endregion

            return index;
        }

        public bool UpdateMessage(DataMessageInternal message, int index)
        {
            if (messageList[index] == null)
                Thread.MemoryBarrier();
            messageList[index].MergeMessage(message);
            return true;
        }

        public void EvaluateQuery(ISession session, QueryMessage message)
        {
            bool lockTaken = false;
            int maxCount = 0;
            QueryMessageDecorator query = new QueryMessageDecorator(session, message);

            addLock.Enter(ref lockTaken);
            maxCount = Count;
            _queryList.Add( query );
            addLock.Exit(true);

            Parallel.For(0, maxCount, index => messageList[index].EvaluateQuery(query));
            session.Flush();
        }


        internal void RemoveQuery(ISession session, QueryMessage message)
        {
            bool lockTaken = false;
            int maxCount = 0;
            addLock.Enter(ref lockTaken);
            maxCount = Count;
            QueryMessageDecorator decorator = null;
            for (int index = 0; index < _queryList.Count; index++ )
            {
                if (_queryList[index].QueryMessage == message)
                {
                    decorator = _queryList[index];
                    _queryList.RemoveAt(index);
                    break;
                }
            }
            addLock.Exit(true);
            if (decorator == null)
                Parallel.For(0, maxCount, index => messageList[index].RemoveQuery(decorator));
        }
    }
}
